package pub.tbc.rubbish.client.discovery.zk;

import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import pub.tbc.rubbish.client.Constant;
import pub.tbc.rubbish.client.discovery.ServiceConnection;
import pub.tbc.rubbish.client.discovery.cache.ServiceCache;
import pub.tbc.rubbish.client.factory.ObjectFactory;
import pub.tbc.rubbish.client.util.ZkHelper;
import pub.tbc.toolkit.Objs;
import pub.tbc.toolkit.PropsUtil;

import java.util.List;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

/**
 * @author tbc on 2016/12/12 18:05:23.
 */
@Slf4j
public class ZKService implements ServiceConnection {
    private CuratorFramework client = ZkHelper.getCuratorFrameworkWithOptions();

    @Override
    public List<String> getChildrenForPath(String serviceName) {
        List<String> children = Lists.newArrayList();
        try {
            children = client.getChildren().forPath(getServicePath(serviceName));
        } catch (Exception e) {
            log.error("获取服务地址失败！");
        }
        if (Objs.isEmpty(children))
            throw new RuntimeException("未获取到服务地址");
        return children.stream().map(this::getServiceAddress).collect(Collectors.toList());
    }

    @Override
    public void listener(String serviceName) {
        String servicePath = getServicePath(serviceName);

        /* 监听子节点的变化情况 */
        PathChildrenCache childrenCache = new PathChildrenCache(
                client,
                servicePath,
                true // 是否缓存数据，如果为true,在收到节点数据变更通知的时候，也能收到节点的数据内容，否则要自己去提取数据；
        );
        try {
            childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 实际维持通信是在另一个线程中的(也可手动给出一个线程池)，此处不用管，不会阻塞当前线程
        childrenCache.getListenable().addListener(
                new PathChildrenCacheListener() {
                    @Override
                    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event)
                            throws Exception {
                        log.info("event type : {}", event.getType());
                        ServiceCache cache = ObjectFactory.getServiceCache();

                        switch (event.getType()) {
                            case CHILD_ADDED:
                                childrenEvent(event, cache::add);
                                break;
                            case CHILD_REMOVED:
                                childrenEvent(event, cache::remove);
                                break;
                            case CHILD_UPDATED:
                                // update 事件，数据改变？
                                break;
                            default:
                                break;
                        }
                    }
                }
        );
    }

    private String getServicePath(String serviceName) {
        String registryNode = PropsUtil.getString(
                PropsUtil.load(Constant.DEFAULT_CONFIG_FILE),
                Constant.REGISTRY_NODE,
                Constant.DEFAULT_REGISTRY_NODE
        );
        return registryNode + "/" + serviceName;
    }

    private String getServiceName(String path) {
        return getElement(path, "/", 2);
    }

    private String getServiceAddress(String path) {
        return getElement(path, "/", 1);
    }

    /**
     * 字符串拆成数组，获取数组的倒数第几个元素
     *
     * @param path  。
     * @param index 。
     * @return 。
     */
    private String getElement(String path, String divide, int index) {
        if (Objs.isEmpty(path))
            return null;
        String[] ss = path.split(divide);
        return ss[ss.length - index];
    }

    private void childrenEvent(PathChildrenCacheEvent event, BiConsumer<String, String> biConsumer) {
        log.info("event path : {}", event.getData().getPath());
        // 发生事件的节点及从节点中提取服务名和地址
        String path = event.getData().getPath();

        // 从path中获取服务名和地址
        String serviceName = getServiceName(path);
        String address = getServiceAddress(path);

//        ServiceCache cache = ObjectFactory.getServiceCache();

        biConsumer.accept(serviceName, address);
    }


}
